package com.neusoft.databus.core.job.impl;

import java.util.List;
import java.util.Optional;

import javax.transaction.Transactional;

import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import com.google.common.eventbus.AsyncEventBus;
import com.neusoft.bizcore.schedule.service.JobMgrService;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.dto.GatewayDTO;
import com.neusoft.databus.common.dto.MetricGroupDTO;
import com.neusoft.databus.common.dto.ResourceDTO;
import com.neusoft.databus.core.event.EventWrapper;
import com.neusoft.databus.core.event.GatewayBatchChangedEvent;
import com.neusoft.databus.core.event.GatewayDeleteEvent;
import com.neusoft.databus.core.event.GatewayStatusChangedEvent;
import com.neusoft.databus.core.event.JobRebuildEvent;
import com.neusoft.databus.core.event.ResourceDeleteEvent;
import com.neusoft.databus.core.event.ResourceStatusChangedEvent;
import com.neusoft.databus.core.event.ResourceUpdateEvent;
import com.neusoft.databus.core.job.BuildJobService;
import com.neusoft.databus.core.job.CollectMetricJob;
import com.neusoft.databus.core.service.GatewayService;
import com.neusoft.databus.core.service.ResourceService;

import lombok.extern.slf4j.Slf4j;

/**
 * 构建定时任务服务
 *
 * @author sunchf
 * @date 2018年12月26日 上午11:33:09
 */
@Slf4j
@Component
@Transactional
public class BuildJobServiceImpl implements BuildJobService, CommandLineRunner {
    @Autowired
    private GatewayService gatewayService;
    @Autowired
    private ResourceService resourceService;
    @Autowired
    private JobMgrService jobMgrService;
    @javax.annotation.Resource(name = "modelUpdateEventBus")
    private AsyncEventBus modelUpdateEventBus;
    @Value("${databus.core.node: 0}")
    private int node;

    //    @PostConstruct
    //    public void init() {
    //        this.modelUpdateEventBus.register(this);
    //    }

    //    @Subscribe
    //    public void onModelUpdateEvent(JobRebuildEvent event) {
    //        this.build(event);
    //    }

    @Override
    public void run(String... args) throws Exception {
        this.build();
    }

    @Override
    public void receiveRedisMessage(String message) {
        BuildJobServiceImpl.log.info("receive a gateway or resource changed redis message: {}", message);
        final EventWrapper wrapper = JsonUtils.jsonToPojo(message, EventWrapper.class);
        JobRebuildEvent event = null;
        switch (wrapper.getEventType()) {
        case gatewayBatchChanged:
            event = JsonUtils.jsonToPojo(wrapper.getEventStr(), GatewayBatchChangedEvent.class);
            break;
        case gatewayDelete:
            event = JsonUtils.jsonToPojo(wrapper.getEventStr(), GatewayDeleteEvent.class);
            break;
        case gatewayStatusChanged:
            event = JsonUtils.jsonToPojo(wrapper.getEventStr(), GatewayStatusChangedEvent.class);
            break;
        case resourceDelete:
            event = JsonUtils.jsonToPojo(wrapper.getEventStr(), ResourceDeleteEvent.class);
            break;
        case resourceStatusChanged:
            event = JsonUtils.jsonToPojo(wrapper.getEventStr(), ResourceStatusChangedEvent.class);
            break;
        case resourceUpdate:
            event = JsonUtils.jsonToPojo(wrapper.getEventStr(), ResourceUpdateEvent.class);
            break;
        default:
            break;
        }

        if (null != event) {
            final String gatewayCode = event.getGateway();
            final GatewayDTO gateway = this.gatewayService.findByCode(gatewayCode);
            if (this.node == gateway.getNode()) {
                this.build(event);
            }
        }
    }

    @Override
    public void build() {
        BuildJobServiceImpl.log.info("begin to build jobs.");
        final long startAt = System.currentTimeMillis();
        final List<GatewayDTO> gateways = this.gatewayService.findAllEnabledAndPulled(this.node);
        for (final GatewayDTO gateway : gateways) {
            if (gateway.isBatched()) {
                final List<ResourceDTO> resources = this.resourceService.findByGateway(gateway.getCode());
                final Optional<Long> minInterval =
                        resources.stream().filter(it -> !it.isDisabled()).map(it -> it.getExecutionCycle())
                                .min((o1, o2) -> o1 > o2 ? 1 : -1);
                if (minInterval.isPresent()) {
                    this.createJob(gateway.getCode(), gateway.getCode(), minInterval.get());
                }
            } else {
                final List<ResourceDTO> resources = this.resourceService.findByGateway(gateway.getCode());
                resources.stream().filter(it -> !it.isDisabled()).forEach(it -> {
                    final MetricGroupDTO mg = it.getCollectedMetricGroup();
                    if ((null != mg) && (mg.getMetricSize() > 0)) {
                        this.createJob(it.getCode(), gateway.getCode(), it.getExecutionCycle());
                    }
                });
            }
        }
        BuildJobServiceImpl.log.info("end to build jobs, elapsed {} ms", System.currentTimeMillis() - startAt);
    }

    @Override
    public void build(JobRebuildEvent event) {

        if (event instanceof GatewayStatusChangedEvent) {
            this.build((GatewayStatusChangedEvent) event);
        } else if (event instanceof GatewayBatchChangedEvent) {
            this.build((GatewayBatchChangedEvent) event);
        } else if (event instanceof GatewayDeleteEvent) {
            this.build((GatewayDeleteEvent) event);
        } else if (event instanceof ResourceUpdateEvent) {
            this.build((ResourceUpdateEvent) event);
        } else if (event instanceof ResourceDeleteEvent) {
            this.build((ResourceDeleteEvent) event);
        } else if (event instanceof ResourceStatusChangedEvent) {
            this.build((ResourceStatusChangedEvent) event);
        }
    }

    /**
     * 当网关删除时，触发任务变更处理
     *
     * @param event
     */
    private void build(GatewayDeleteEvent event) {
        //删除相同网关批量采集任务(如果存在的话)
        this.removeJob(event.getGateway(), event.getGateway());
        //删除单个资源采集任务(如果存在的话)
        event.getResources().forEach(resource -> {
            this.removeJob(resource, event.getGateway());
        });
    }

    /**
     * 当网关状态发生变化时，触发任务变更处理
     *
     * @param event
     */
    private void build(GatewayStatusChangedEvent event) {
        if (event.isBatched()) {
            if (event.isDisabled()) {
                this.removeJob(event.getGateway(), event.getGateway());
            } else {
                //创建一个新的同网关批量采集任务
                final Optional<Long> optMinInterval =
                        event.getResources().values().stream().min((o1, o2) -> o1 > o2 ? 1 : -1);
                if (optMinInterval.isPresent()) {
                    this.createJob(event.getGateway(), event.getGateway(), optMinInterval.get());
                } else {
                    this.createJob(event.getGateway(), event.getGateway(), 60 * 3);
                }
            }
        } else {
            if (event.isDisabled()) {
                event.getResources().entrySet().forEach(entry -> {
                    this.removeJob(entry.getKey(), event.getGateway());
                });
            } else {
                event.getResources().entrySet().forEach(entry -> {
                    this.createJob(entry.getKey(), event.getGateway(), entry.getValue());
                });
            }
        }
    }

    /**
     * 当网关批处理发生变化时，触发任务变更处理
     *
     * @param event
     */
    private void build(GatewayBatchChangedEvent event) {

        if (event.isBatched()) {
            event.getResources().entrySet().forEach(entry -> {
                this.removeJob(entry.getKey(), event.getGateway());
            });
            //创建一个新的同网关批量采集任务
            final Optional<Long> optMinInterval =
                    event.getResources().values().stream().min((o1, o2) -> o1 > o2 ? 1 : -1);
            if (optMinInterval.isPresent()) {
                this.createJob(event.getGateway(), event.getGateway(), optMinInterval.get());
            } else {
                this.createJob(event.getGateway(), event.getGateway(), 60 * 3);
            }

        } else {
            this.removeJob(event.getGateway(), event.getGateway());
            event.getResources().entrySet().forEach(entry -> {
                this.createJob(entry.getKey(), event.getGateway(), entry.getValue());
            });
        }
    }

    /**
     * 当资源更新时，触发任务变更处理 <br>
     * 事件发送方已经控制相关逻辑，收到此事件，说明需要创建或更新任务<br >
     *
     * @param event
     */
    private void build(ResourceUpdateEvent event) {
        this.createJob(event.getResource(), event.getGateway(), event.getInterval());
    }

    /**
     * 当资源删除时，触发任务变更处理
     *
     * @param event
     */
    private void build(ResourceDeleteEvent event) {
        this.removeJob(event.getResource(), event.getGateway());
    }

    /**
     * 当资源禁用/启用时，触发任务变更处理
     *
     * @param event
     */
    private void build(ResourceStatusChangedEvent event) {
        if (event.isDisabled()) {
            this.removeJob(event.getResource(), event.getGateway());
        } else {
            this.createJob(event.getResource(), event.getGateway(), event.getInterval());
        }
    }

    private void createJob(String name, String group, long interval) {
        final boolean existed = this.jobMgrService.isExists(name, group);
        if (existed) {
            this.jobMgrService.rescheduleJob(name, group, interval);
        } else {
            final String jobDesc = name.equals(group) ? "相同网关批量合并任务,网关编码：" + name : "单个资源采集任务，资源编码：" + name;
            final JobDetail jobDetail =
                    JobBuilder.newJob(CollectMetricJob.class).withIdentity(name, group)
                            .withDescription(jobDesc).build();
            this.jobMgrService.scheduleJob(name, group, interval, jobDetail);
        }
    }

    private void removeJob(String name, String group) {
        final boolean existed = this.jobMgrService.isExists(name, group);
        if (existed) {
            this.jobMgrService.pauseJob(name, group);
            this.jobMgrService.deleteJob(name, group);
        }
    }

}
